# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: helpers\issue_tracker\clients\tfs_client.py
import json, html
from . import util
from ..exceptions import ConfigError
from .base_client import BaseClient
from requests_ntlm import HttpNtlmAuth


class TFSClient(BaseClient):
    """
    TFSClient handles all communications between Acunetix and a user-specified
     Microsoft TFS instance.
    
    All communication is done through Microsoft's TFS API version 2.0 and does
     not currently provide any backwards compatibility to TFS API version 1.0
     or any possible future versions just yet.
    
    Version 2.0 of Microsoft TFS' API supports the following authentication methods:
         (a). NTLM (Implemented)
         (b). Basic (TODO?)
         (c). OAuth (Not Implementing)
    
    :Example:
    
    >>> import helpers.issue_tracker.clients.client as client
    >>>
    >>> target_config = {
    >>>     "bug_tracker": "tfs",
    >>>         "url": "http://tfs.local:8080/tfs////",
    >>>         "auth": {
    >>>             "kind": "ntlm",
    >>>             "user": "root",
    >>>             "password": "toor"
    >>>         },
    >>>         "project": 'acx/acx',
    >>>         "issuetype": "bug"  # Optional
    >>> }
    >>> client = client.get_client()
    """

    def __init__(self, target_config, proxies_cb=None, ssl_ca_bundle_path=True):
        super().__init__(target_config, proxies_cb, ssl_ca_bundle_path)

    def _format_vulnerability_description(self, description):
        pass

    def _format_vulnerability_references(self, references):
        html_references = []
        try:
            for pair in references:
                if not isinstance(pair, (list, tuple)):
                    raise AssertionError("Invalid type, must be of type list or tuple")
                if len(pair) == 1:
                    html_references.append("<a href=" + pair[0] + "></a>")
                elif len(pair) == 2:
                    html_references.append(
                        "<a href=" + pair[1] + ">" + pair[0] + "</a>"
                    )
                else:
                    raise ValueError(
                        "Reference pair has too many or too few items to unpack, "
                        + "must be in pairs of 1 or 2"
                    )

        except (AttributeError, TypeError):
            raise AssertionError("References must be an iterable object")

        return html_references

    def _request(self, **kwargs):
        try:
            return super()._request(**kwargs)
        except AttributeError:
            raise RuntimeError(
                "Attempting to parse data when no user credentials have been saved"
            )

    def _establish_session(self):
        if self.target_config["auth"]["kind"] == "ntlm":
            collection = (
                self.target_config.get("project", {})
                .get("project_name", "DefaultCollection")
                .split("/")[0]
            )
            request = self._request(
                endpoint="/" + collection + "/_apis/projects?api-version=2.0",
                auth=HttpNtlmAuth(self.username, self.password),
            )
            if request.status_code != 200:
                raise ConfigError(
                    "Invalid username and password or account locked out."
                )
            try:
                if not request.headers.get("X-TFS-Session"):
                    if request.headers.get("X-TFS-ProcessId"):
                        if request.headers.get("ActivityId"):
                            if request.headers.get("X-VSS-UserData"):
                                raise KeyError
            except KeyError:
                raise ConfigError(
                    "Unexpected response returned. Normally caused by an invalid URL."
                )

        else:
            raise ConfigError("Unknown Authentication method parsed - Must be 'ntlm'.")

    def _get_collections(self):
        request = self._request(
            endpoint="/_apis/projectcollections",
            auth=HttpNtlmAuth(self.username, self.password),
        )
        if request.status_code == 200:
            try:
                return [collection["name"] for collection in request.json()["value"]]
            except KeyError:
                raise Exception(
                    "Unable to parse response body for list of collections for this TFS instance."
                )

            raise Exception(
                "Unable to fetch list of collections for this TFS instance."
            )

    def add_vulnerability(self, vuln_data):
        """Create and push a new issue to a TFS instance with vulnerability data.
        
        Do note that TFS names issues as WorkItems; instead of containing a description
         it has a field know as ReproSteps or Reproduction Steps. The following method
         will post the vulnerability with the vulnerability title and add the description
         inside the ReproSteps in the following format:
        
          # Vulnerability Severity
          ## Vulnerability Description
           ----
          ## Attack Details
           ----
          ## References
        
        :Example:
        
        >>> import helpers.issue_tracker.clients.client as client
        >>>
        >>> # Sample data provided by Acunetix
        >>> vuln_data = {
        >>>    'vt_name': 'foo',
        >>>    'long_description': '',
        >>>    'description': 'Test Description',
        >>>    'details': 'Attack Details',
        >>>    'vt_tags': ['sql_injection', 'CWE-89']
        >>> }
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "tfs",
        >>>         "url": "http://tfs.local:8080/tfs////",
        >>>         "auth": {
        >>>             "kind": "ntlm",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>> }
        >>>
        >>> client = client.get_client(target_config)
        >>> client.add_vulnerability(vuln_data)
        
        :note: Vulnerability description is formatted automatically by the client and should
        not be modified in any way prior (only pass it as raw HTML).
        
        :param vuln_data: WorkItem data added into TFS. Format:
                            {
                                'op': 'add',
                                'path': '/fields/System.Title',
                                'value': 'Acunetix - ' + vuln_data['vt_name']
                            },
                            {
                                'op': 'add',
                                'path': '/fields/Microsoft.VSTS.TCM.ReproSteps',
                                'value': vulnerability_description
                            }
        :return: New Issue ID if successfully created (201), otherwise None.
        """
        repo = self.target_config["project"]["project_name"]
        request_data = vuln_data.get("request")
        if request_data:
            request_details = "<pre>" + request_data + "</pre>"
        else:
            request_details = ""
        target_information = (
            '<table style="border:1px solid #ccc"><tr>'
            + '<td style="border-right:1px solid #ccc;border-bottom:1px solid #ccc; '
            + 'padding-right: 5px;"><strong>Target URL</strong></td><td style="border-bottom:1px '
            + 'solid #ccc">'
            + vuln_data.get("target_address")
            + '</td></tr><tr><td style="border-right:1px solid #ccc;"><strong>Severity</strong></td><td>'
            + util.evaluate_severity(vuln_data.get("severity"))
            + "</td></tr></table>"
        )
        recommendation = (
            vuln_data["recommendation"] if vuln_data["recommendation"] else ""
        )
        impact = vuln_data["impact"] if vuln_data["impact"] else ""
        vulnerability_description = (
            target_information
            + "<h3>Vulnerability Description</h3>"
            + vuln_data["description"]
            + "<br/><hr><h3>Attack Details</h3>"
            + vuln_data["details"]
            + "<br/><h3>HTTP Request</h3>"
            + request_details
            + "<br/><h3>Impact</h3>"
            + impact
            + "<br/><h3>Remediation</h3>"
            + recommendation
        )
        reference_tags = self._format_vulnerability_references(vuln_data["refs"])
        if len(reference_tags) > 0:
            vulnerability_description += "<br/><hr><h5>References</h5><br/>"
            for reference_tag in reference_tags:
                vulnerability_description += reference_tag + "<br/>"

            vulnerability_description += "<br/>"
        vulnerability_description = html.unescape(vulnerability_description)
        list_of_tags = super()._format_vulnerability_tags(vuln_data["vt_tags"])
        if "issue_type" in self.target_config.keys():
            list_of_tags.append(self.target_config["issue_type"]["issue_id"])
        tags = ""
        for tag in list_of_tags:
            tags += tag + ";"

        tags = tags[:-1] if len(tags) > 0 else ""
        post_data = json.dumps(
            [
                {
                    "op": "add",
                    "path": "/fields/System.Title",
                    "value": "Acunetix - " + vuln_data["vt_name"],
                },
                {
                    "op": "add",
                    "path": "/fields/Microsoft.VSTS.TCM.ReproSteps",
                    "value": vulnerability_description,
                },
                {"op": "add", "path": "/fields/System.Tags", "value": tags},
            ]
        ).encode("utf-8")
        request = self.session.patch(
            url=self.target_config["url"]
            + "/"
            + repo
            + "/_apis/wit/workItems/$Bug?api-version=2.0",
            headers={"Content-Type": "application/json-patch+json"},
            data=post_data,
            auth=HttpNtlmAuth(self.username, self.password),
            timeout=BaseClient.timeout,
            verify=False,
        )
        try:
            if request.status_code == 200:
                return request.json()["id"]
            raise Exception(
                "Invalid status code returned. Unable to add issue successfully."
            )
        except KeyError:
            raise Exception(
                "Unexpected response body returned, unable to parse issue number."
            )

    def get_projects(self):
        """Get list of projects available to authenticated user.
        
        >>> import helpers.issue_tracker.clients.client as client
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "tfs",
        >>>         "url": "http://tfs.local:8080/tfs////",
        >>>         "auth": {
        >>>             "kind": "ntlm",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>> }
        >>>
        >>> client = client.get_client(target_config)
        >>> client.get_projects()
        >>> ['DefaultProject']
        
        :return: List of projects available for the authenticated user
        """
        collection = (
            self.target_config.get("project", {})
            .get("collection", "DefaultCollection")
            .split("/")[0]
        )
        request = self._request(
            endpoint="/" + collection + "/_apis/projects?api-version=2.0",
            auth=HttpNtlmAuth(self.username, self.password),
        )
        if request.status_code == 200:
            try:
                projects = request.json()["value"]
                return [
                    {
                        "project_id": str(project["id"]),
                        "project_name": "DefaultCollection/" + project["name"],
                    }
                    for project in projects
                ]
            except KeyError:
                raise Exception(
                    "Received unexpected response body, unable to parse project name and ID."
                )

            raise Exception(
                "Request to TFS unsuccessful. Unable to retrieve list of projects."
            )

    def get_issue_types(self):
        """Extract issue-types for each project.
        
        .. note::
        
        Format is returned in { "projects": [
                                    "PROJECT_NAME": {
                                        "issuetypes": [
                                        ]
                                    }
                                 ]
                              }
        
        :return: List of issues types per project for authenticated user.
        """
        repository = self.target_config["project"]["project_name"]
        request = self._request(
            endpoint="/" + repository + "/_apis/wit/workItemTypes?api-version=2.0",
            auth=HttpNtlmAuth(self.username, self.password),
        )
        if request.status_code == 200:
            issue_types = [
                {"issue_id": issuetype["name"], "issue_name": issuetype["name"]}
                for issuetype in request.json()["value"]
            ]
        else:
            issue_types = None
        return issue_types

    def get_issue_status(self, issue_id):
        """Get the status of a specific issue (e.g. In Progress, Done)
        
        >>> import helpers.issue_tracker.clients.client as client
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "tfs",
        >>>         "url": "http://tfs.local:8080/tfs////",
        >>>         "auth": {
        >>>             "kind": "ntlm",
        >>>             "user": "root",
        >>>             "password": "toor"
        >>>         },
        >>>         "project": 'acx/acx',
        >>>         "issuetype": "bug"  # Optional
        >>> }
        >>>
        >>> client = client.get_client(target_config)
        >>> client.get_issue_status(1)
        >>> 'Bug'
        >>> client.get_issue_status(-1)
        >>> None
        
        :param issue_id: ID of issue to extract status from
        :return: Status of the specified issue
        """
        collection = (
            self.target_config.get("project", {})
            .get("project_name", "DefaultCollection")
            .split("/")[0]
        )
        request = self._request(
            endpoint="/"
            + collection
            + "/_apis/wit/workitems/"
            + str(issue_id)
            + "?api-version=2.0",
            auth=HttpNtlmAuth(self.username, self.password),
        )
        try:
            if request.status_code == 200:
                return request.json()["fields"]["System.WorkItemType"]
        except KeyError:
            raise Exception("Unexpected response body returned")

        raise Exception("Unable to retrieve issue status for issue: " + str(issue_id))
